/**
 * Created by Administrator on 2025/5/7.
 * */
#include "AIPC.h"
#include <iostream>
#include <thread>
#include <sys/socket.h>
#include <arpa/inet.h>
#include <unistd.h>

BaseObject *createAIPC() {
    return new AIPC();
}

void destroyAIPC(BaseObject *p) {
    SAFE_DELETE(p);
}

/*__global__ void TaskKernel(char *in, char *out) {
    int tid = blockIdx.x * blockDim.x + threadIdx.x;
    out[tid] = in[tid];
}*/

AIPC::AIPC(){
    this->setGPUDevice(0);
    prevStep = -1;
    // 订阅数据源主题Topic
    sub = zmq::socket_t(context, zmq::socket_type::sub);
    sub.connect(topicAddr.c_str());
    sub.set(zmq::sockopt::subscribe, topicName);
    // 初始化ZMQ服务端
    rep_socket = zmq::socket_t(context, zmq::socket_type::rep);
    rep_socket.bind(rep_address);
    std::thread thd(&AIPC::zmqRepServe, this);
    thd.detach();
}

AIPC::~AIPC() {
}

void AIPC::zmqRepServe() {
    // 处理CFAR控制命令
    zmq::message_t command;
    try {
        rep_socket.recv(command, zmq::recv_flags::none);
    } catch (const zmq::error_t& e) {
        std::cerr << "ZMQ error: " << e.what() << std::endl;
    }
    int cmdId;
    memcpy(&cmdId, command.data(), sizeof(int));
    if (1 == cmdId) {
        std::cout << "AINodeInfer::AIPC：开始模型升级" << std::endl;
    }
    // 发送确认
    int ackVal = 200;
    zmq::message_t ack(sizeof(int));
    memcpy(ack.data(), &ackVal, sizeof(int));
    rep_socket.send(ack, zmq::send_flags::none);
}

void AIPC::onIssueStreamedCudaOperations(buffer_table_t *input, buffer_table_t *output, cudaStream_t stream) {
    int step;
    zmq::message_t msgTopicName;
    zmq::message_t msgData;
    if (!sub.recv(msgTopicName, zmq::recv_flags::none)) {
        std::cerr << "Subscribe: Failed to receive topic name" << std::endl;
    }
    std::string topicName_(static_cast<char*>(msgTopicName.data()), msgTopicName.size());
    if (!sub.recv(msgData, zmq::recv_flags::none)) {
        std::cerr << "Subscribe: Failed to receive data" << std::endl;
    }
    memcpy(&step, msgData.data(), sizeof(float));
    if (prevStep == step) {
        return ;
    }
    prevStep = step;
    // 根据全局aiPCState确定相应的操作
    int aiPCState = GlobalParameters::shareInstance()->get_parameter_aiPCState(dynamic_cast<TaskModule*>(this));
    if (aiPCState==static_cast<int>(AIPCStateEnum::Ini_Model_Notify)) {
        std::cout << "AINodeInfer::AIPC: 开始升级模型..." << std::endl;
        auto aiPCStateHolder = GlobalParameters::shareInstance()->findParameter("aiPCState");
        int *aiPCState = static_cast<int*>(GlobalParameters::shareInstance()->getValue(aiPCStateHolder, dynamic_cast<TaskModule*>(this)));
        *aiPCState = static_cast<int>(AIPCStateEnum::Ini_Model_Updating);
    }
    if (aiPCState==static_cast<int>(AIPCStateEnum::Ini_Model_Updating) && step==STEP4_1) {
        std::cout << "AINodeInfer::AIPC: 升级模型成功" << std::endl;
        inferTime = INFER_TIME2;
        auto aiPCStateHolder = GlobalParameters::shareInstance()->findParameter("aiPCState");
        int *aiPCState = static_cast<int*>(GlobalParameters::shareInstance()->getValue(aiPCStateHolder, dynamic_cast<TaskModule*>(this)));
        *aiPCState = static_cast<int>(AIPCStateEnum::Ini_Model_Updated);
    }
}
